package simpledb.index;

import simpledb.common.Database;
import simpledb.common.DbException;
import simpledb.common.Debug;
import simpledb.common.Permissions;
import simpledb.execution.IndexPredicate;
import simpledb.execution.Predicate.Op;
import simpledb.storage.*;
import simpledb.transaction.TransactionAbortedException;
import simpledb.transaction.TransactionId;

import java.io.*;
import java.util.*;

/**
 * BTreeFile 是一个实现了 DbFile 接口的类，用于存储 B+ 树索引结构。
 *
 * 该类负责管理：
 * - 指向根节点页面的指针（BTreeRootPtrPage）
 * - 一组内部节点页面（BTreeInternalPage）
 * - 一组叶子节点页面（BTreeLeafPage），其中按顺序存储了元组数据
 *
 * BTreeFile 与 BTreeLeafPage、BTreeInternalPage 和 BTreeRootPtrPage 紧密协作，
 * 页面格式详见这些类的构造函数。
 */
public class BTreeFile implements DbFile {

    private final File      f;       // 文件对象，用于磁盘存储
    private final TupleDesc td;      // 元组描述信息
    private final int       tableid; // 表唯一标识符
    private final int       keyField; // 用作索引键的字段索引

    /**
     * 构造方法：创建一个由指定文件支持的 B+ 树文件。
     *
     * @param f   - 存储此 B+ 树文件的磁盘文件
     * @param key - 作为索引键的字段索引
     * @param td  - 文件中元组的描述信息
     */
    public BTreeFile(File f, int key, TupleDesc td) {
        this.f = f;
        this.tableid = f.getAbsoluteFile().hashCode(); // 使用文件绝对路径哈希生成表 ID
        this.keyField = key;
        this.td = td;
    }

    /**
     * 返回与此 BTreeFile 对应的磁盘文件对象。
     */
    public File getFile() {
        return f;
    }

    /**
     * 返回此 BTreeFile 唯一标识符（tableid）。
     * 实现建议：使用文件绝对路径的哈希值。
     *
     * @return 此 BTreeFile 的唯一 ID
     */
    public int getId() {
        return tableid;
    }

    /**
     * 获取此数据库文件中存储的元组描述信息。
     *
     * @return TupleDesc 类型描述信息
     */
    public TupleDesc getTupleDesc() {
        return td;
    }

    /**
     * 从磁盘读取指定页面内容并构造为 Page 对象。
     * 不应直接调用，而应通过 BufferPool.getPage() 调用。
     *
     * @param pid - 要读取的页面 ID
     * @return 从磁盘读取后构造的 Page 对象
     */
    public Page readPage(PageId pid) {
        BTreePageId id = (BTreePageId) pid;

        try (BufferedInputStream bis = new BufferedInputStream(new FileInputStream(f))) {
            if (id.pgcateg() == BTreePageId.ROOT_PTR) {
                byte[] pageBuf = new byte[BTreeRootPtrPage.getPageSize()];
                int retval = bis.read(pageBuf, 0, BTreeRootPtrPage.getPageSize());
                if (retval == -1) {
                    throw new IllegalArgumentException("读取超出文件末尾");
                }
                if (retval < BTreeRootPtrPage.getPageSize()) {
                    throw new IllegalArgumentException("无法从 BTreeFile 中读取完整的根指针页面");
                }
                Debug.log(1, "BTreeFile.readPage: 读取页面 %d", id.getPageNumber());
                return new BTreeRootPtrPage(id, pageBuf);
            } else {
                byte[] pageBuf = new byte[BufferPool.getPageSize()];
                long offset = BTreeRootPtrPage.getPageSize() + (long) (id.getPageNumber() - 1)
                              * BufferPool.getPageSize();
                if (bis.skip(offset) != offset) {
                    throw new IllegalArgumentException("无法定位到 BTreeFile 中的正确位置");
                }
                int retval = bis.read(pageBuf, 0, BufferPool.getPageSize());
                if (retval == -1) {
                    throw new IllegalArgumentException("读取超出文件末尾");
                }
                if (retval < BufferPool.getPageSize()) {
                    throw new IllegalArgumentException("无法从 BTreeFile 中读取完整的页面");
                }
                Debug.log(1, "BTreeFile.readPage: 读取页面 %d", id.getPageNumber());

                if (id.pgcateg() == BTreePageId.INTERNAL) {
                    return new BTreeInternalPage(id, pageBuf, keyField);
                } else if (id.pgcateg() == BTreePageId.LEAF) {
                    return new BTreeLeafPage(id, pageBuf, keyField);
                } else { // HEADER 类型
                    return new BTreeHeaderPage(id, pageBuf);
                }
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        // 成功或失败后关闭文件流
    }

    /**
     * 将页面写入磁盘。不应直接调用，应由 BufferPool 在刷新页面时调用。
     *
     * @param page - 要写入磁盘的页面
     */
    public void writePage(Page page) throws IOException {
        BTreePageId id = (BTreePageId) page.getId();

        byte[] data = page.getPageData();
        RandomAccessFile rf = new RandomAccessFile(f, "rw");

        if (id.pgcateg() == BTreePageId.ROOT_PTR) {
            rf.write(data); // 根页面位于文件开头
            rf.close();
        } else {
            long offset = BTreeRootPtrPage.getPageSize() + (long) (page.getId().getPageNumber() - 1)
                          * BufferPool.getPageSize();
            rf.seek(offset); // 定位到对应页面偏移位置
            rf.write(data); // 写入页面数据
            rf.close();
        }
    }

    /**
     * 返回此 BTreeFile 中包含的页面总数。
     */
    public int numPages() {
        // 只有完整页面会被写入
        return (int) ((f.length() - BTreeRootPtrPage.getPageSize()) / BufferPool.getPageSize());
    }

    /**
     * 返回当前 B+ 树使用的索引键字段索引。
     */
    public int keyField() {
        return keyField;
    }

    /**
     * 递归查找并锁定目标叶子页面。沿路径对所有内部节点加 READ_ONLY 锁，
     * 最终对目标叶子节点加上指定权限的锁。
     * <p>
     * 如果 f 为 null，则查找最左侧叶子页面 —— 用于迭代器扫描整个文件。
     *
     * @param tid        - 事务 ID
     * @param dirtypages - 脏页列表，用于记录新产生的脏页
     * @param pid        - 当前正在搜索的页面 ID
     * @param perm       - 给叶子页面加的锁权限
     * @param f          - 要查找的字段值
     * @return 找到的可能包含字段 f 的最左叶子页面
     */
    private BTreeLeafPage findLeafPage(TransactionId tid, Map<PageId, Page> dirtypages, BTreePageId pid,
                                       Permissions perm, Field f) throws DbException, TransactionAbortedException {

        // 如果是叶子页面，直接返回
        if (pid.pgcateg() == BTreePageId.LEAF) {
            return (BTreeLeafPage) getPage(tid, dirtypages, pid, perm);
        }

        // 否则当前页面是内部页面，获取其迭代器
        BTreeInternalPage internalPage = (BTreeInternalPage) getPage(tid, dirtypages, pid, Permissions.READ_ONLY);
        Iterator<BTreeEntry> iterator = internalPage.iterator();
        BTreeEntry entry = null;

        while (iterator.hasNext()) {
            entry = iterator.next();
            if (f == null) {
                // 如果 f 为 null，递归查找最左子节点
                return findLeafPage(tid, dirtypages, entry.getLeftChild(), perm, f);
            }
            if (entry.getKey().compare(Op.GREATER_THAN_OR_EQ, f)) {
                // 如果当前键 >= f，进入左子节点
                return findLeafPage(tid, dirtypages, entry.getLeftChild(), perm, f);
            }
        }

        // 如果没有找到合适的 entry，进入最后一个 entry 的右子节点
        return entry == null ? null : findLeafPage(tid, dirtypages, entry.getRightChild(), perm, f);
    }

    /**
     * 便捷方法：在不传递 dirtypages 的情况下查找叶子页面。
     * 主要用于 BTreeFile 的迭代器。
     *
     * @param tid - 事务 ID
     * @param pid - 当前搜索的页面 ID
     * @param f   - 要查找的字段值
     * @return 找到的可能包含字段 f 的最左叶子页面
     */
    BTreeLeafPage findLeafPage(TransactionId tid, BTreePageId pid, Field f) throws DbException,
                                                                           TransactionAbortedException {
        return findLeafPage(tid, new HashMap<>(), pid, Permissions.READ_ONLY, f);
    }

    /**
     * 拆分一个叶子页面以容纳新元组，并递归地拆分父节点（如有需要）以容纳新的条目。
     * 新的条目应具有与右页第一个元组相同的键（即“上提”），并指向拆分后的两个叶子页。
     * 需要更新兄弟指针和父指针。
     * <p>
     * 返回应该插入带有指定 key 字段的新元组的叶子页。
     *
     * @param tid        - 事务 ID
     * @param dirtypages - 脏页列表，用于记录所有新增的脏页
     * @param page       - 要拆分的叶子页
     * @param field      - 插入完成后要插入的新元组的 key 字段。必须知道这个字段，
     *                   才能决定返回左页还是右页
     * @return 应该插入新元组的叶子页
     * @throws DbException
     * @throws IOException
     * @throws TransactionAbortedException
     * @see #getParentWithEmptySlots(TransactionId, Map, BTreePageId, Field)
     */
    public BTreeLeafPage splitLeafPage(TransactionId tid, Map<PageId, Page> dirtypages, BTreeLeafPage page, Field field)
                                                                                                                        throws DbException,
                                                                                                                        IOException,
                                                                                                                        TransactionAbortedException {
        // 创建一个新的叶子页
        final BTreeLeafPage newPage = (BTreeLeafPage) getEmptyPage(tid, dirtypages, BTreePageId.LEAF);

        // 计算需要移动多少个元组到新页（当前页的一半）
        int moved = page.getNumTuples() / 2;

        // 使用反向迭代器从后往前取出元组
        final Iterator<Tuple> iterator = page.reverseIterator();
        final List<Tuple> movedTuples = new ArrayList<>();
        Tuple next = null;
        while (iterator.hasNext() && moved > 0) {
            moved--;
            next = iterator.next();
            page.deleteTuple(next); // 删除当前页中的元组
            movedTuples.add(next); // 添加到待移动列表中
        }

        // 确保至少有一个元组被移动（防止 mid 元组为 null）
        assert next != null : "The mid leaf entry should not be null";

        // 将移动的元组按原顺序插入新页（注意倒序处理）
        for (int i = movedTuples.size() - 1; i >= 0; i--) {
            newPage.insertTuple(movedTuples.get(i));
        }

        // 更新兄弟页面指针（如果当前页有右兄弟）
        if (page.getRightSiblingId() != null) {
            final BTreePageId rightId = page.getRightSiblingId();
            final BTreeLeafPage rightPage = (BTreeLeafPage) getPage(tid, dirtypages, rightId, Permissions.READ_ONLY);
            rightPage.setLeftSiblingId(newPage.getId()); // 设置右兄弟的左兄弟为新页
            newPage.setRightSiblingId(rightPage.getId()); // 设置新页的右兄弟为原右兄弟
            dirtypages.put(rightPage.getId(), rightPage); // 标记右兄弟为脏页
        }

        // 更新当前页和新页的兄弟指针
        newPage.setLeftSiblingId(page.getId()); // 新页的左兄弟是当前页
        page.setRightSiblingId(newPage.getId()); // 当前页的右兄弟是新页
        dirtypages.put(page.getId(), page); // 标记当前页为脏页
        dirtypages.put(newPage.getId(), newPage); // 标记新页为脏页

        // 获取合适的父节点内部页（确保有空槽位插入新条目）
        final BTreeInternalPage parentPage = getParentWithEmptySlots(tid, dirtypages, page.getParentId(), field);

        // 构造新条目：key 是新页的第一个元组的 key，左右孩子分别是原页和新页
        final BTreeEntry entry = new BTreeEntry(next.getField(parentPage.keyField), page.getId(), newPage.getId());

        // 在父页中插入新条目
        parentPage.insertEntry(entry);
        dirtypages.put(parentPage.getId(), parentPage); // 标记父页为脏页

        // 更新子页的父指针
        updateParentPointer(tid, dirtypages, parentPage.getId(), page.getId());
        updateParentPointer(tid, dirtypages, parentPage.getId(), newPage.getId());

        // 判断插入方向：根据 field 和 mid key 比较结果决定返回哪一页
        final boolean isRight = field.compare(Op.GREATER_THAN_OR_EQ, next.getField(parentPage.keyField));
        return isRight ? newPage : page;
    }

    /**
     * 拆分一个内部页以容纳新条目，并递归地拆分其父页（如有需要）以容纳新的条目。
     * 新条目的 key 应该等于原始内部页的中间 key（即“上推”），child 指针指向拆分后的两个内部页。
     * 需要更新所有迁移到新页的子节点的父指针。
     * <p>
     * 返回应该插入带有指定 key 字段的新条目的内部页。
     *
     * @param tid        - 事务 ID
     * @param dirtypages - 脏页列表，用于记录所有新增的脏页
     * @param page       - 要拆分的内部页
     * @param field      - 插入完成后要插入的新条目的 key 字段。必须知道这个字段，
     *                   才能决定返回左页还是右页
     * @return 应该插入新条目的内部页
     * @throws DbException
     * @throws IOException
     * @throws TransactionAbortedException
     * @see #getParentWithEmptySlots(TransactionId, Map, BTreePageId, Field)
     * @see #updateParentPointers(TransactionId, Map, BTreeInternalPage)
     */
    public BTreeInternalPage splitInternalPage(TransactionId tid, Map<PageId, Page> dirtypages, BTreeInternalPage page,
                                               Field field) throws DbException, IOException,
                                                           TransactionAbortedException {
        // 创建一个新的内部页
        final BTreeInternalPage newPage = (BTreeInternalPage) getEmptyPage(tid, dirtypages, BTreePageId.INTERNAL);

        // 计算需要移动多少个条目到新页（当前页的一半）
        int moved = page.getNumEntries() / 2;

        // 使用反向迭代器从后往前取出条目
        final Iterator<BTreeEntry> iterator = page.reverseIterator();
        final List<BTreeEntry> movedEntries = new ArrayList<>();
        BTreeEntry next = null;
        while (iterator.hasNext() && moved > 0) {
            moved--;
            next = iterator.next();
            page.deleteKeyAndRightChild(next); // 删除当前页中的条目
            movedEntries.add(next); // 添加到待移动列表
        }

        // 将移动的条目按原顺序插入新页（注意倒序处理）
        for (int i = movedEntries.size() - 1; i >= 0; i--) {
            newPage.insertEntry(movedEntries.get(i));
        }

        // 取出中间条目（下一个可用的条目）
        final BTreeEntry mid = iterator.next();
        page.deleteKeyAndRightChild(mid); // 删除中间条目

        // 设置中间条目的左右子页
        mid.setLeftChild(page.getId());
        mid.setRightChild(newPage.getId());

        // 更新所有子节点的父指针
        updateParentPointers(tid, dirtypages, page);
        updateParentPointers(tid, dirtypages, newPage);

        // 标记当前页和新页为脏页
        dirtypages.put(page.getId(), page);
        dirtypages.put(newPage.getId(), newPage);

        // 获取合适的父节点内部页（确保有空槽位插入新条目）
        final BTreeInternalPage parentPage = getParentWithEmptySlots(tid, dirtypages, page.getParentId(), field);

        // 在父页中插入中间条目
        parentPage.insertEntry(mid);
        dirtypages.put(parentPage.getId(), parentPage); // 标记父页为脏页

        // 判断插入方向：根据 field 和 mid key 比较结果决定返回哪一页
        final boolean isRight = field.compare(Op.GREATER_THAN_OR_EQ, mid.getKey());
        return isRight ? newPage : page;
    }

    /**
     * 封装获取一个可以容纳新条目的父页的过程。
     * 可能需要创建一个新的内部页作为树的新根；
     * 如果当前父页没有空槽位，则进行分裂；
     * 否则直接锁定并返回现有的父页。
     *
     * @param tid        - 事务 ID
     * @param dirtypages - 脏页列表，用于记录所有新增的脏页
     * @param parentId   - 父页的 ID。可能是内部页或 RootPtr 页
     * @param field      - 要插入的条目 key。如果父页需要分裂时需要用到这个 key
     * @return 保证至少有一个空槽位的父页
     * @throws DbException
     * @throws IOException
     * @throws TransactionAbortedException
     * @see #splitInternalPage(TransactionId, Map, BTreeInternalPage, Field)
     */
    private BTreeInternalPage getParentWithEmptySlots(TransactionId tid, Map<PageId, Page> dirtypages,
                                                      BTreePageId parentId, Field field) throws DbException,
                                                                                        IOException,
                                                                                        TransactionAbortedException {

        BTreeInternalPage parent = null;

        // 如果是 ROOT_PTR 页面（说明还没有内部页），需要创建新的根节点
        if (parentId.pgcateg() == BTreePageId.ROOT_PTR) {
            parent = (BTreeInternalPage) getEmptyPage(tid, dirtypages, BTreePageId.INTERNAL);

            // 获取并更新 root 指针页面
            BTreeRootPtrPage rootPtr = (BTreeRootPtrPage) getPage(tid, dirtypages, BTreeRootPtrPage.getId(tableid),
                Permissions.READ_WRITE);
            BTreePageId prevRootId = rootPtr.getRootId(); // 保存旧的根 ID
            rootPtr.setRootId(parent.getId()); // 设置新根

            // 更新旧根的父指针为新创建的内部页（即新根）
            BTreePage prevRootPage = (BTreePage) getPage(tid, dirtypages, prevRootId, Permissions.READ_WRITE);
            prevRootPage.setParentId(parent.getId());
        } else {
            // 否则直接获取已有的父页（读写锁）
            parent = (BTreeInternalPage) getPage(tid, dirtypages, parentId, Permissions.READ_WRITE);
        }

        // 如果父页没有空槽位，就进行分裂
        if (parent.getNumEmptySlots() == 0) {
            parent = splitInternalPage(tid, dirtypages, parent, field);
        }

        return parent;
    }

    /**
     * 辅助函数：更新某个节点的父指针。
     *
     * @param tid        - 事务 ID
     * @param dirtypages - 脏页列表
     * @param pid        - 父节点的 ID
     * @param child      - 子节点的 ID，需要设置其父指针
     * @throws DbException
     * @throws TransactionAbortedException
     */
    private void updateParentPointer(TransactionId tid, Map<PageId, Page> dirtypages, BTreePageId pid, BTreePageId child)
                                                                                                                         throws DbException,
                                                                                                                         TransactionAbortedException {

        BTreePage p = (BTreePage) getPage(tid, dirtypages, child, Permissions.READ_ONLY);

        // 如果子页的父指针不是指定的父页 ID，则更新
        if (!p.getParentId().equals(pid)) {
            p = (BTreePage) getPage(tid, dirtypages, child, Permissions.READ_WRITE);
            p.setParentId(pid);
        }
    }

    /**
     * 更新给定页的所有子页的父指针，使其正确指向该页。
     *
     * @param tid        - 事务 ID
     * @param dirtypages - 脏页列表
     * @param page       - 父页对象
     * @throws DbException
     * @throws TransactionAbortedException
     * @see #updateParentPointer(TransactionId, Map, BTreePageId, BTreePageId)
     */
    private void updateParentPointers(TransactionId tid, Map<PageId, Page> dirtypages, BTreeInternalPage page)
                                                                                                              throws DbException,
                                                                                                              TransactionAbortedException {
        Iterator<BTreeEntry> it = page.iterator();
        BTreePageId pid = page.getId();
        BTreeEntry e = null;
        while (it.hasNext()) {
            e = it.next();
            updateParentPointer(tid, dirtypages, pid, e.getLeftChild());
        }
        if (e != null) {
            updateParentPointer(tid, dirtypages, pid, e.getRightChild());
        }
    }

    /**
     * 封装获取/锁定页面的方法。
     * 首先检查本地缓存 "dirtypages"，若找不到再从缓冲池中获取。
     * 如果是以读写权限获取的页面，会加入到 dirtypages 缓存中（因为可能会被修改）。
     * <p>
     * 该方法用于确保多次访问同一页面时不会丢失更新。
     *
     * @param tid        - 事务 ID
     * @param dirtypages - 脏页列表
     * @param pid        - 请求的页面 ID
     * @param perm       - 请求的页面权限（READ_ONLY / READ_WRITE）
     * @return 请求的页面
     * @throws DbException
     * @throws IOException
     * @throws TransactionAbortedException
     */
    Page getPage(TransactionId tid, Map<PageId, Page> dirtypages, BTreePageId pid, Permissions perm)
                                                                                                    throws DbException,
                                                                                                    TransactionAbortedException {
        if (dirtypages.containsKey(pid)) {
            return dirtypages.get(pid);
        } else {
            Page p = Database.getBufferPool().getPage(tid, pid, perm);
            if (perm == Permissions.READ_WRITE) {
                dirtypages.put(pid, p);
            }
            return p;
        }
    }

    /**
     * 插入一个元组到这个 B+ 树文件中，保持元组按 key 字段排序。
     * 如果目标叶子页已满，可能引起页面分裂。
     *
     * @param tid - 事务 ID
     * @param t   - 要插入的元组
     * @return 本次操作所“脏化”的所有页面列表
     * @see #splitLeafPage(TransactionId, Map, BTreeLeafPage, Field)
     */
    public List<Page> insertTuple(TransactionId tid, Tuple t) throws DbException, IOException,
                                                             TransactionAbortedException {
        Map<PageId, Page> dirtypages = new HashMap<>();

        // 获取 root pointer page 并定位 root page
        BTreeRootPtrPage rootPtr = getRootPtrPage(tid, dirtypages);
        BTreePageId rootId = rootPtr.getRootId();

        // 如果 rootId 为空，说明这是第一次插入，需要初始化 root page
        if (rootId == null) {
            rootId = new BTreePageId(tableid, numPages(), BTreePageId.LEAF);
            rootPtr = (BTreeRootPtrPage) getPage(tid, dirtypages, BTreeRootPtrPage.getId(tableid),
                Permissions.READ_WRITE);
            rootPtr.setRootId(rootId); // 设置根页
        }

        // 查找应该插入的叶子页，并加写锁
        BTreeLeafPage leafPage = findLeafPage(tid, dirtypages, rootId, Permissions.READ_WRITE, t.getField(keyField));

        // 如果叶子页已满，进行分裂
        if (leafPage.getNumEmptySlots() == 0) {
            leafPage = splitLeafPage(tid, dirtypages, leafPage, t.getField(keyField));
        }

        // 插入元组
        leafPage.insertTuple(t);

        // 返回所有脏页
        return new ArrayList<>(dirtypages.values());
    }

    /**
     * 当一个 B+ 树页面由于删除操作导致占用率低于半满时进行处理。
     * 如果该页的某个兄弟页有“多余”的条目，则从兄弟页中“借”一些；
     * 否则就与兄弟页合并，并更新相应的指针。
     *
     * @param tid        - 事务 ID
     * @param dirtypages - 脏页列表，记录所有被修改的页面
     * @param page       - 占用率低于半满的页面
     * @throws DbException
     * @throws IOException
     * @throws TransactionAbortedException
     * @see #handleMinOccupancyLeafPage(TransactionId, Map, BTreeLeafPage, BTreeInternalPage, BTreeEntry, BTreeEntry)
     * @see #handleMinOccupancyInternalPage(TransactionId, Map, BTreeInternalPage, BTreeInternalPage, BTreeEntry, BTreeEntry)
     */
    private void handleMinOccupancyPage(TransactionId tid, Map<PageId, Page> dirtypages, BTreePage page)
                                                                                                        throws DbException,
                                                                                                        IOException,
                                                                                                        TransactionAbortedException {
        BTreePageId parentId = page.getParentId();
        BTreeEntry leftEntry = null;
        BTreeEntry rightEntry = null;
        BTreeInternalPage parent = null;

        // 如果当前页不是根页，则找到它的父页
        if (parentId.pgcateg() != BTreePageId.ROOT_PTR) {
            parent = (BTreeInternalPage) getPage(tid, dirtypages, parentId, Permissions.READ_WRITE);
            Iterator<BTreeEntry> ite = parent.iterator();

            // 遍历父页中的条目，找到指向当前页及其左右兄弟页的条目
            while (ite.hasNext()) {
                BTreeEntry e = ite.next();
                if (e.getLeftChild().equals(page.getId())) {
                    rightEntry = e; // 右边的兄弟
                    break;
                } else if (e.getRightChild().equals(page.getId())) {
                    leftEntry = e; // 左边的兄弟
                }
            }
        }

        // 如果是叶子页，调用处理叶子页下溢的方法
        if (page.getId().pgcateg() == BTreePageId.LEAF) {
            handleMinOccupancyLeafPage(tid, dirtypages, (BTreeLeafPage) page, parent, leftEntry, rightEntry);
        } else { // 内部页
            handleMinOccupancyInternalPage(tid, dirtypages, (BTreeInternalPage) page, parent, leftEntry, rightEntry);
        }
    }

    /**
     * 处理叶子页由于删除导致占用率不足的情况。
     * 如果有一个兄弟页有多余的元组，就从中“借”；
     * 否则就与兄弟页合并，并更新指针。
     *
     * @param tid        - 事务 ID
     * @param dirtypages - 脏页列表
     * @param page       - 当前叶子页（占用率不足）
     * @param parent     - 当前页的父页
     * @param leftEntry  - 父页中指向当前页及其左兄弟的条目
     * @param rightEntry - 父页中指向当前页及其右兄弟的条目
     * @throws DbException
     * @throws IOException
     * @throws TransactionAbortedException
     * @see #mergeLeafPages(TransactionId, Map, BTreeLeafPage, BTreeLeafPage, BTreeInternalPage, BTreeEntry)
     * @see #stealFromLeafPage(BTreeLeafPage, BTreeLeafPage, BTreeInternalPage, BTreeEntry, boolean)
     */
    private void handleMinOccupancyLeafPage(TransactionId tid, Map<PageId, Page> dirtypages, BTreeLeafPage page,
                                            BTreeInternalPage parent, BTreeEntry leftEntry, BTreeEntry rightEntry)
                                                                                                                  throws DbException,
                                                                                                                  IOException,
                                                                                                                  TransactionAbortedException {

        BTreePageId leftSiblingId = null;
        BTreePageId rightSiblingId = null;

        if (leftEntry != null)
            leftSiblingId = leftEntry.getLeftChild(); // 获取左兄弟页 ID
        if (rightEntry != null)
            rightSiblingId = rightEntry.getRightChild(); // 获取右兄弟页 ID

        int maxEmptySlots = page.getMaxTuples() - page.getMaxTuples() / 2; // 计算半满阈值（向上取整）

        // 尝试从左兄弟“借”元组
        if (leftSiblingId != null) {
            BTreeLeafPage leftSibling = (BTreeLeafPage) getPage(tid, dirtypages, leftSiblingId, Permissions.READ_WRITE);

            // 如果左兄弟也处于最小占用状态，则合并；否则“借”一些元组
            if (leftSibling.getNumEmptySlots() >= maxEmptySlots) {
                mergeLeafPages(tid, dirtypages, leftSibling, page, parent, leftEntry);
            } else {
                stealFromLeafPage(page, leftSibling, parent, leftEntry, false);
            }
        }
        // 尝试从右兄弟“借”
        else if (rightSiblingId != null) {
            BTreeLeafPage rightSibling = (BTreeLeafPage) getPage(tid, dirtypages, rightSiblingId,
                Permissions.READ_WRITE);

            // 如果右兄弟也处于最小占用状态，则合并；否则“借”一些元组
            if (rightSibling.getNumEmptySlots() >= maxEmptySlots) {
                mergeLeafPages(tid, dirtypages, page, rightSibling, parent, rightEntry);
            } else {
                stealFromLeafPage(page, rightSibling, parent, rightEntry, true);
            }
        }
    }

    /**
     * 从兄弟页中“借”一些元组，使当前页恢复到至少半满状态。
     * 同时更新父页中的对应条目，使其 key 值为右子页的第一个元组的 key。
     *
     * @param page           - 当前页（占用率不足）
     * @param sibling        - 兄弟页（有多余元组）
     * @param parent         - 父页
     * @param entry          - 父页中连接这两个叶子页的条目
     * @param isRightSibling - 是否是从右兄弟“借”
     * @throws DbException
     */
    public void stealFromLeafPage(BTreeLeafPage page, BTreeLeafPage sibling, BTreeInternalPage parent,
                                  BTreeEntry entry, boolean isRightSibling) throws DbException {

        // 当前页和兄弟页的元组数
        int curTuples = page.getNumTuples();
        int siblingTuples = sibling.getNumTuples();

        // 目标元组数：平均分配
        int targetTuples = (curTuples + siblingTuples) >> 1;

        // 根据是左兄弟还是右兄弟，选择迭代方向
        Iterator<Tuple> it = isRightSibling ? sibling.iterator() : sibling.reverseIterator();

        // 搬移元组
        while (it.hasNext() && curTuples < targetTuples) {
            Tuple t = it.next();
            sibling.deleteTuple(t); // 从兄弟页删除
            page.insertTuple(t); // 插入当前页
            curTuples++;
        }

        // 更新父页中的 key，为兄弟页的第一个元组的 key
        Tuple mid = sibling.iterator().next();
        entry.setKey(mid.getField(parent.keyField));
        parent.updateEntry(entry);
    }

    /**
     * 处理由于删除操作导致内部页（非叶子页）占用率低于半满的情况。
     * 如果该页的某个兄弟页有“多余”的条目，则从兄弟页中“借”一些；
     * 否则就与兄弟页合并，并更新相应的指针。
     *
     * @param tid        - 事务 ID
     * @param dirtypages - 脏页列表，记录所有被修改的页面
     * @param page       - 当前内部页（占用率不足）
     * @param parent     - 当前页的父页
     * @param leftEntry  - 父页中指向当前页及其左兄弟的条目
     * @param rightEntry - 父页中指向当前页及其右兄弟的条目
     * @throws DbException
     * @throws IOException
     * @throws TransactionAbortedException
     * @see #mergeInternalPages(TransactionId, Map, BTreeInternalPage, BTreeInternalPage, BTreeInternalPage, BTreeEntry)
     * @see #stealFromLeftInternalPage(TransactionId, Map, BTreeInternalPage, BTreeInternalPage, BTreeInternalPage, BTreeEntry)
     * @see #stealFromRightInternalPage(TransactionId, Map, BTreeInternalPage, BTreeInternalPage, BTreeInternalPage, BTreeEntry)
     */
    private void handleMinOccupancyInternalPage(TransactionId tid, Map<PageId, Page> dirtypages,
                                                BTreeInternalPage page, BTreeInternalPage parent, BTreeEntry leftEntry,
                                                BTreeEntry rightEntry) throws DbException, IOException,
                                                                      TransactionAbortedException {

        // 获取左右兄弟页 ID
        BTreePageId leftSiblingId = null;
        BTreePageId rightSiblingId = null;
        if (leftEntry != null)
            leftSiblingId = leftEntry.getLeftChild();
        if (rightEntry != null)
            rightSiblingId = rightEntry.getRightChild();

        // 计算最小占用阈值（即最大空槽数量为总容量的一半向上取整）
        int maxEmptySlots = page.getMaxEntries() - page.getMaxEntries() / 2;

        // 尝试从左兄弟借条目
        if (leftSiblingId != null) {
            BTreeInternalPage leftSibling = (BTreeInternalPage) getPage(tid, dirtypages, leftSiblingId,
                Permissions.READ_WRITE);

            // 判断是否可以借用（兄弟页是否有富余）
            if (leftSibling.getNumEmptySlots() >= maxEmptySlots) {
                mergeInternalPages(tid, dirtypages, leftSibling, page, parent, leftEntry); // 合并
            } else {
                stealFromLeftInternalPage(tid, dirtypages, page, leftSibling, parent, leftEntry); // 借用
            }
        }
        // 尝试从右兄弟借条目
        else if (rightSiblingId != null) {
            BTreeInternalPage rightSibling = (BTreeInternalPage) getPage(tid, dirtypages, rightSiblingId,
                Permissions.READ_WRITE);

            if (rightSibling.getNumEmptySlots() >= maxEmptySlots) {
                mergeInternalPages(tid, dirtypages, page, rightSibling, parent, rightEntry); // 合并
            } else {
                stealFromRightInternalPage(tid, dirtypages, page, rightSibling, parent, rightEntry); // 借用
            }
        }
    }

    /**
     * 从左兄弟内部页中“借”一些条目，使当前页恢复到至少半满状态。
     * 条目可以看作是通过父条目“旋转”的方式移动：
     * - 父条目的 key 被“拉下”到当前页；
     * - 左兄弟的最后一个 key 被“推上”到父条目；
     * 更新所有涉及子节点的父指针。
     *
     * @param tid         - 事务 ID
     * @param dirtypages  - 脏页列表
     * @param page        - 当前内部页（占用率不足）
     * @param leftSibling - 左兄弟页（有多余条目）
     * @param parent      - 当前页和兄弟页的父页
     * @param parentEntry - 父页中连接这两个内部页的条目
     * @throws DbException
     * @throws TransactionAbortedException
     * @see #updateParentPointers(TransactionId, Map, BTreeInternalPage)
     */
    public void stealFromLeftInternalPage(TransactionId tid, Map<PageId, Page> dirtypages, BTreeInternalPage page,
                                          BTreeInternalPage leftSibling, BTreeInternalPage parent,
                                          BTreeEntry parentEntry) throws DbException, TransactionAbortedException {

        // 目标条目数：平均分配
        int moved = (leftSibling.getNumEntries() + page.getNumEntries()) >> 1;

        final Iterator<BTreeEntry> iterator = leftSibling.reverseIterator(); // 反向遍历左兄弟

        // 1. 将父条目的 key 拉下来作为新 entry 插入当前页
        final BTreeEntry right = iterator.next(); // 取左兄弟最后一个 entry
        leftSibling.deleteKeyAndRightChild(right); // 删除该 entry 的右子节点

        final BTreeEntry left = page.iterator().next(); // 当前页第一个 entry
        final BTreeEntry entry = new BTreeEntry(parentEntry.getKey(), right.getRightChild(), left.getLeftChild());
        page.insertEntry(entry); // 插入到当前页
        page.insertEntry(right); // 插入左兄弟的最后一个 entry

        // 2. 继续搬移条目直到达到目标数量
        int curTuples = page.getNumEntries();
        while (curTuples < moved && iterator.hasNext()) {
            final BTreeEntry next = iterator.next();
            leftSibling.deleteKeyAndRightChild(next);
            page.insertEntry(next);
            curTuples++;
        }

        // 3. 更新父条目的 key：使用左兄弟剩余的第一个 entry 的 key
        final BTreeEntry mid = iterator.next();
        leftSibling.deleteKeyAndRightChild(mid);
        parentEntry.setKey(mid.getKey());
        parent.updateEntry(parentEntry);

        // 更新所有子节点的父指针
        updateParentPointers(tid, dirtypages, page);
    }

    /**
     * 从右兄弟内部页中“借”一些条目，使当前页恢复到至少半满状态。
     * 条目可以看作是通过父条目“旋转”的方式移动：
     * - 父条目的 key 被“拉下”到当前页；
     * - 右兄弟的第一个 key 被“推上”到父条目；
     * 更新所有涉及子节点的父指针。
     *
     * @param tid          - 事务 ID
     * @param dirtypages   - 脏页列表
     * @param page         - 当前内部页（占用率不足）
     * @param rightSibling - 右兄弟页（有多余条目）
     * @param parent       - 当前页和兄弟页的父页
     * @param parentEntry  - 父页中连接这两个内部页的条目
     * @throws DbException
     * @throws TransactionAbortedException
     * @see #updateParentPointers(TransactionId, Map, BTreeInternalPage)
     */
    public void stealFromRightInternalPage(TransactionId tid, Map<PageId, Page> dirtypages, BTreeInternalPage page,
                                           BTreeInternalPage rightSibling, BTreeInternalPage parent,
                                           BTreeEntry parentEntry) throws DbException, TransactionAbortedException {

        // 目标条目数：平均分配
        int moved = (rightSibling.getNumEntries() + page.getNumEntries()) >> 1;

        final Iterator<BTreeEntry> iterator = rightSibling.iterator(); // 正向遍历右兄弟

        // 1. 将父条目的 key 拉下来作为新 entry 插入当前页
        BTreeEntry right = iterator.next(); // 取右兄弟第一个 entry
        rightSibling.deleteKeyAndLeftChild(right); // 删除该 entry 的左子节点

        BTreeEntry left = page.reverseIterator().next(); // 当前页最后一个 entry
        BTreeEntry entry = new BTreeEntry(parentEntry.getKey(), left.getRightChild(), right.getLeftChild());
        page.insertEntry(entry); // 插入到当前页
        page.insertEntry(right); // 插入右兄弟的第一个 entry

        // 2. 继续搬移条目直到达到目标数量
        int curTuples = page.getNumEntries();
        while (curTuples < moved && iterator.hasNext()) {
            final BTreeEntry next = iterator.next();
            rightSibling.deleteKeyAndLeftChild(next);
            page.insertEntry(next);
            curTuples++;
        }

        // 3. 更新父条目的 key：使用右兄弟剩余的第一个 entry 的 key
        final BTreeEntry mid = iterator.next();
        rightSibling.deleteKeyAndLeftChild(mid);
        parentEntry.setKey(mid.getKey());
        parent.updateEntry(parentEntry);

        // 更新所有子节点的父指针
        updateParentPointers(tid, dirtypages, page);
    }

    /**
     * 合并两个叶子页，将右页的所有元组移动到左页。
     * 删除父页中对应的 key 和右子指针，并递归处理父页下溢的情况。
     * 更新兄弟页指针，并将右页标记为可重用。
     *
     * @param tid         - 事务 ID
     * @param dirtypages  - 脏页列表，记录所有被修改的页面
     * @param leftPage    - 左边的叶子页
     * @param rightPage   - 右边的叶子页
     * @param parent      - 两页的父内部页
     * @param parentEntry - 父页中指向这两个叶子页的条目
     * @throws DbException
     * @throws IOException
     * @throws TransactionAbortedException
     * @see #deleteParentEntry(TransactionId, Map, BTreePage, BTreeInternalPage, BTreeEntry)
     */
    public void mergeLeafPages(TransactionId tid, Map<PageId, Page> dirtypages, BTreeLeafPage leftPage,
                               BTreeLeafPage rightPage, BTreeInternalPage parent, BTreeEntry parentEntry)
                                                                                                         throws DbException,
                                                                                                         IOException,
                                                                                                         TransactionAbortedException {

        // 将右页中的所有元组移动到左页中
        final Iterator<Tuple> iterator = rightPage.iterator();
        while (iterator.hasNext()) {
            final Tuple next = iterator.next();
            rightPage.deleteTuple(next); // 从右页删除
            leftPage.insertTuple(next); // 插入左页
        }

        // 更新兄弟页指针
        final BTreePageId rrId = rightPage.getRightSiblingId(); // 获取右页的右兄弟
        if (rrId != null) {
            // 如果有右兄弟，更新其左兄弟为左页
            final BTreeLeafPage rr = (BTreeLeafPage) getPage(tid, dirtypages, rrId, Permissions.READ_WRITE);
            rr.setLeftSiblingId(leftPage.getId());
            leftPage.setRightSiblingId(rrId); // 左页右兄弟设为右兄弟
            dirtypages.put(rrId, rr); // 记录修改
        } else {
            // 如果没有右兄弟，左页右兄弟为空
            leftPage.setRightSiblingId(null);
        }

        // 标记右页为空页（可用于重用）
        setEmptyPage(tid, dirtypages, rightPage.pid.getPageNumber());

        // 删除父页中的对应条目，并处理父页可能的下溢
        deleteParentEntry(tid, dirtypages, leftPage, parent, parentEntry);

        // 将修改后的左页加入脏页列表
        dirtypages.put(leftPage.getId(), leftPage);
    }

    /**
     * 合并两个内部页，将右页的所有条目移动到左页，
     * 并“拉下”父条目中的 key 到左页中作为分隔键。
     * 删除父页中对应的 key 和右子指针，并递归处理父页下溢。
     * 更新子节点的父指针，并将右页标记为可重用。
     *
     * @param tid         - 事务 ID
     * @param dirtypages  - 脏页列表
     * @param leftPage    - 左边的内部页
     * @param rightPage   - 右边的内部页
     * @param parent      - 两页的父内部页
     * @param parentEntry - 父页中指向这两个内部页的条目
     * @throws DbException
     * @throws IOException
     * @throws TransactionAbortedException
     * @see #deleteParentEntry(TransactionId, Map, BTreePage, BTreeInternalPage, BTreeEntry)
     * @see #updateParentPointers(TransactionId, Map, BTreeInternalPage)
     */
    public void mergeInternalPages(TransactionId tid, Map<PageId, Page> dirtypages, BTreeInternalPage leftPage,
                                   BTreeInternalPage rightPage, BTreeInternalPage parent, BTreeEntry parentEntry)
                                                                                                                 throws DbException,
                                                                                                                 IOException,
                                                                                                                 TransactionAbortedException {

        // 1. 拉下父条目的 key，并插入一个新 entry 连接左右页
        BTreeEntry lastLeftEntry = leftPage.reverseIterator().next(); // 左页最后一个 entry
        BTreeEntry firstRightEntry = rightPage.iterator().next(); // 右页第一个 entry

        BTreeEntry newEntry = new BTreeEntry(parentEntry.getKey(), // 父条目的 key
            lastLeftEntry.getRightChild(), // 左页最后 entry 的右孩子
            firstRightEntry.getLeftChild() // 右页第一个 entry 的左孩子
        );
        leftPage.insertEntry(newEntry); // 插入连接 entry

        // 2. 将右页中的所有 entry 移动到左页
        Iterator<BTreeEntry> it = rightPage.iterator();
        while (it.hasNext()) {
            BTreeEntry e = it.next();
            rightPage.deleteKeyAndLeftChild(e); // 删除右页中的 entry
            leftPage.insertEntry(e); // 插入左页
        }

        // 3. 更新所有子节点的父指针
        updateParentPointers(tid, dirtypages, leftPage);

        // 4. 清空右页，使其可被重用
        setEmptyPage(tid, dirtypages, rightPage.pid.getPageNumber());

        // 5. 删除父页中的对应条目，并处理父页可能的下溢
        deleteParentEntry(tid, dirtypages, leftPage, parent, parentEntry);
    }

    /**
     * 封装从父节点中删除条目（key + 右子指针）的过程。
     * 如果父节点变空，说明它是根节点，应被替换为其唯一子节点。
     * 否则如果它低于最小占用率，则尝试借用或合并。
     *
     * @param tid         - 事务 ID
     * @param dirtypages  - 脏页列表
     * @param leftPage    - 删除后保留的子页
     * @param parent      - 包含要删除条目的父页
     * @param parentEntry - 要删除的条目
     * @throws DbException
     * @throws IOException
     * @throws TransactionAbortedException
     * @see #handleMinOccupancyPage(TransactionId, Map, BTreePage)
     */
    private void deleteParentEntry(TransactionId tid, Map<PageId, Page> dirtypages, BTreePage leftPage,
                                   BTreeInternalPage parent, BTreeEntry parentEntry) throws DbException, IOException,
                                                                                    TransactionAbortedException {

        // 删除父页中的条目
        parent.deleteKeyAndRightChild(parentEntry);

        int maxEmptySlots = parent.getMaxEntries() - parent.getMaxEntries() / 2; // 最小占用阈值

        // 如果父页为空了，说明是根页，需要替换为子页
        if (parent.getNumEmptySlots() == parent.getMaxEntries()) {
            BTreePageId rootPtrId = parent.getParentId();
            if (rootPtrId.pgcateg() != BTreePageId.ROOT_PTR) {
                throw new DbException("attempting to delete a non-root node");
            }

            BTreeRootPtrPage rootPtr = (BTreeRootPtrPage) getPage(tid, dirtypages, rootPtrId, Permissions.READ_WRITE);
            leftPage.setParentId(rootPtrId); // 设置新根页的父指针
            rootPtr.setRootId(leftPage.getId()); // 更新根页 ID

            // 释放旧根页
            setEmptyPage(tid, dirtypages, parent.getId().getPageNumber());
        }
        // 如果父页下溢了，继续处理
        else if (parent.getNumEmptySlots() > maxEmptySlots) {
            handleMinOccupancyPage(tid, dirtypages, parent);
        }
    }

    /**
     * 从 B+ 树文件中删除一个元组。
     * 如果删除导致页的使用率低于半满，可能会触发合并或重新分配操作。
     *
     * @param tid - 事务 ID
     * @param t   - 要删除的元组
     * @return 所有在此次操作中被修改的页（即脏页）
     * @see #handleMinOccupancyPage(TransactionId, Map, BTreePage)
     */
    public List<Page> deleteTuple(TransactionId tid, Tuple t) throws DbException, IOException,
                                                             TransactionAbortedException {

        Map<PageId, Page> dirtypages = new HashMap<>();

        // 定位元组所在的叶子页
        BTreePageId pageId = new BTreePageId(tableid, t.getRecordId().getPageId().getPageNumber(), BTreePageId.LEAF);
        BTreeLeafPage page = (BTreeLeafPage) getPage(tid, dirtypages, pageId, Permissions.READ_WRITE);
        page.deleteTuple(t); // 删除元组

        // 判断是否下溢
        int maxEmptySlots = page.getMaxTuples() - page.getMaxTuples() / 2;
        if (page.getNumEmptySlots() > maxEmptySlots) {
            handleMinOccupancyPage(tid, dirtypages, page); // 触发下溢处理
        }

        return new ArrayList<>(dirtypages.values());
    }

    /**
     * 获取对根指针页的读锁。如果不存在，则创建根指针页和根页。
     *
     * @param tid        - 事务 ID
     * @param dirtypages - 脏页列表，记录所有被修改的页面
     * @return 根指针页（BTreeRootPtrPage）
     * @throws DbException
     * @throws IOException
     * @throws TransactionAbortedException
     */
    BTreeRootPtrPage getRootPtrPage(TransactionId tid, Map<PageId, Page> dirtypages) throws DbException, IOException,
                                                                                    TransactionAbortedException {

        synchronized (this) {
            // 如果文件长度为0，表示这是第一次初始化
            if (f.length() == 0) {
                // 创建空的根指针页和一个空的叶子页作为初始结构
                BufferedOutputStream bw = new BufferedOutputStream(new FileOutputStream(f, true));
                byte[] emptyRootPtrData = BTreeRootPtrPage.createEmptyPageData();
                byte[] emptyLeafData = BTreeLeafPage.createEmptyPageData();
                bw.write(emptyRootPtrData); // 写入根指针页
                bw.write(emptyLeafData); // 写入第一个叶子页
                bw.close();
            }
        }

        // 获取根指针页的只读锁并返回
        return (BTreeRootPtrPage) getPage(tid, dirtypages, BTreeRootPtrPage.getId(tableid), Permissions.READ_ONLY);
    }

    /**
     * 获取该 B+ 树文件中的第一个空闲页号。
     * 如果没有空闲页存在，则新建一页。
     *
     * @param tid        - 事务 ID
     * @param dirtypages - 脏页列表，记录所有被修改的页面
     * @return 第一个空闲页的页号
     * @throws DbException
     * @throws IOException
     * @throws TransactionAbortedException
     */
    public int getEmptyPageNo(TransactionId tid, Map<PageId, Page> dirtypages) throws DbException, IOException,
                                                                              TransactionAbortedException {

        // 获取根指针页以定位到 Header 页链表
        BTreeRootPtrPage rootPtr = getRootPtrPage(tid, dirtypages);
        BTreePageId headerId = rootPtr.getHeaderId();

        int emptyPageNo = 0;

        // 遍历 Header 页链表，查找是否有可用空页槽
        if (headerId != null) {
            BTreeHeaderPage headerPage = (BTreeHeaderPage) getPage(tid, dirtypages, headerId, Permissions.READ_ONLY);
            int headerPageCount = 0;

            while (headerPage != null && headerPage.getEmptySlot() == -1) {
                // 当前 Header 页无空槽，继续找下一个
                headerId = headerPage.getNextPageId();
                if (headerId != null) {
                    headerPage = (BTreeHeaderPage) getPage(tid, dirtypages, headerId, Permissions.READ_ONLY);
                    headerPageCount++;
                } else {
                    headerPage = null;
                }
            }

            // 找到有空槽的 Header 页
            if (headerPage != null) {
                // 切换为可写模式
                headerPage = (BTreeHeaderPage) getPage(tid, dirtypages, headerId, Permissions.READ_WRITE);
                int emptySlot = headerPage.getEmptySlot();
                headerPage.markSlotUsed(emptySlot, true); // 标记为已使用

                // 计算全局页号
                emptyPageNo = headerPageCount * BTreeHeaderPage.getNumSlots() + emptySlot;
            }
        }

        // 如果没有找到 Header 页或 Header 页都满了
        if (headerId == null) {
            synchronized (this) {
                // 新建一个页（默认为内部页格式），扩展文件大小
                BufferedOutputStream bw = new BufferedOutputStream(new FileOutputStream(f, true));
                byte[] emptyData = BTreeInternalPage.createEmptyPageData();
                bw.write(emptyData);
                bw.close();

                // 返回新页的页号（当前总页数）
                emptyPageNo = numPages();
            }
        }

        return emptyPageNo;
    }

    /**
     * 封装创建新页的过程。优先复用旧页，若无可复用则新建一页。
     * 擦除磁盘和缓存中的旧数据，返回一个干净的新页（具有读写锁）。
     *
     * @param tid        - 事务 ID
     * @param dirtypages - 脏页列表
     * @param pgcateg    - 新页类型（LEAF、INTERNAL 或 HEADER）
     * @return 干净的新页对象
     * @throws DbException
     * @throws IOException
     * @throws TransactionAbortedException
     * @see #getEmptyPageNo(TransactionId, Map)
     * @see #setEmptyPage(TransactionId, Map, int)
     */
    private Page getEmptyPage(TransactionId tid, Map<PageId, Page> dirtypages, int pgcateg) throws DbException,
                                                                                           IOException,
                                                                                           TransactionAbortedException {

        // 获取一个空页号
        int emptyPageNo = getEmptyPageNo(tid, dirtypages);

        // 构造新的页 ID
        BTreePageId newPageId = new BTreePageId(tableid, emptyPageNo, pgcateg);

        // 在磁盘上写入空页数据
        RandomAccessFile rf = new RandomAccessFile(f, "rw");
        rf.seek(BTreeRootPtrPage.getPageSize() + (long) (emptyPageNo - 1) * BufferPool.getPageSize());
        rf.write(BTreePage.createEmptyPageData());
        rf.close();

        // 确保缓冲池和脏页列表中没有残留
        Database.getBufferPool().discardPage(newPageId);
        dirtypages.remove(newPageId);

        // 获取新页并加读写锁
        return getPage(tid, dirtypages, newPageId, Permissions.READ_WRITE);
    }

    /**
     * 标记一个页为“空闲”，即可以被复用。
     * 查找或创建对应的 Header 页，并在该 Header 页中标记对应槽位为空。
     *
     * @param tid         - 事务 ID
     * @param dirtypages  - 脏页列表，记录所有被修改的页面
     * @param emptyPageNo - 要标记为空的页号
     * @throws DbException
     * @throws IOException
     * @throws TransactionAbortedException
     * @see #getEmptyPage(TransactionId, Map, int)
     */
    public void setEmptyPage(TransactionId tid, Map<PageId, Page> dirtypages, int emptyPageNo) throws DbException,
                                                                                              IOException,
                                                                                              TransactionAbortedException {

        // 注释掉的部分：如果是最后一个页，就直接截断文件
        // 但为了事务回滚等原因，这部分被禁用了
        BTreeRootPtrPage rootPtr = getRootPtrPage(tid, dirtypages);
        BTreePageId headerId = rootPtr.getHeaderId();
        BTreePageId prevId = null;
        int headerPageCount = 0;

        if (headerId == null) {
            // 获取可写的根指针页
            rootPtr = (BTreeRootPtrPage) getPage(tid, dirtypages, BTreeRootPtrPage.getId(tableid),
                Permissions.READ_WRITE);

            // 创建新 Header 页
            BTreeHeaderPage headerPage = (BTreeHeaderPage) getEmptyPage(tid, dirtypages, BTreePageId.HEADER);
            headerId = headerPage.getId();
            headerPage.init(); // 初始化 Header 页
            rootPtr.setHeaderId(headerId); // 更新根指针页中的 Header 页指针
        }

        while (headerId != null && (headerPageCount + 1) * BTreeHeaderPage.getNumSlots() < emptyPageNo) {
            BTreeHeaderPage headerPage = (BTreeHeaderPage) getPage(tid, dirtypages, headerId, Permissions.READ_ONLY);
            prevId = headerId;
            headerId = headerPage.getNextPageId();
            headerPageCount++;
        }

        while ((headerPageCount + 1) * BTreeHeaderPage.getNumSlots() < emptyPageNo) {
            BTreeHeaderPage prevPage = (BTreeHeaderPage) getPage(tid, dirtypages, prevId, Permissions.READ_WRITE);

            BTreeHeaderPage headerPage = (BTreeHeaderPage) getEmptyPage(tid, dirtypages, BTreePageId.HEADER);
            headerId = headerPage.getId();
            headerPage.init();
            headerPage.setPrevPageId(prevId);
            prevPage.setNextPageId(headerId);

            headerPageCount++;
            prevId = headerId;
        }

        BTreeHeaderPage headerPage = (BTreeHeaderPage) getPage(tid, dirtypages, headerId, Permissions.READ_WRITE);
        int emptySlot = emptyPageNo - headerPageCount * BTreeHeaderPage.getNumSlots();
        headerPage.markSlotUsed(emptySlot, false);
    }

    /**
     * 获取符合索引谓词条件的元组迭代器。
     * 会获取受影响页的读锁，可能会阻塞直到获得锁。
     *
     * @param tid   - 事务 ID
     * @param ipred - 索引谓词
     * @return 元组过滤后的迭代器
     */
    public DbFileIterator indexIterator(TransactionId tid, IndexPredicate ipred) {
        return new BTreeSearchIterator(this, tid, ipred);
    }

    /**
     * 获取按顺序遍历整个 B+ 树文件的元组迭代器。
     * 会获取受影响页的读锁，可能会阻塞直到获得锁。
     *
     * @param tid - 事务 ID
     * @return 整个文件元组的迭代器
     */
    public DbFileIterator iterator(TransactionId tid) {
        return new BTreeFileIterator(this, tid);
    }
}

/**
 * 该类实现了 Java 的 Iterator 接口，用于遍历 B+ 树文件中的所有元组（Tuple）。
 * 它是 AbstractDbFileIterator 的具体实现，能够按顺序访问所有的叶子页数据。
 */
class BTreeFileIterator extends AbstractDbFileIterator {

    // 当前页面的元组迭代器
    Iterator<Tuple>     it   = null;

    // 当前正在读取的叶子页
    BTreeLeafPage       curp = null;

    // 当前事务 ID
    final TransactionId tid;

    // 所属的 B+ 树文件
    final BTreeFile     f;

    /**
     * 构造方法
     * @param f - 包含元组的 BTreeFile 对象
     * @param tid - 当前事务 ID
     */
    public BTreeFileIterator(BTreeFile f, TransactionId tid) {
        this.f = f;
        this.tid = tid;
    }

    /**
     * 打开迭代器，获取第一个叶子页的元组迭代器
     */
    public void open() throws DbException, TransactionAbortedException {
        // 获取根指针页（root pointer page）
        BTreeRootPtrPage rootPtr = (BTreeRootPtrPage) Database.getBufferPool().getPage(tid,
            BTreeRootPtrPage.getId(f.getId()), Permissions.READ_ONLY);

        // 获取根节点页 ID
        BTreePageId root = rootPtr.getRootId();

        // 查找第一个叶子页（不需要谓词，所以参数为 null）
        curp = f.findLeafPage(tid, root, null);

        // 获取该页的元组迭代器
        it = curp.iterator();
    }

    /**
     * 读取下一个元组：
     * 如果当前页还有元组则返回下一个；
     * 否则继续读取右兄弟页。
     *
     * @return 下一个元组，若无更多元组则返回 null
     */
    @Override
    protected Tuple readNext() throws TransactionAbortedException, DbException {
        // 如果当前页已读完，则重置 it 为 null
        if (it != null && !it.hasNext())
            it = null;

        // 循环查找下一个非空页
        while (it == null && curp != null) {
            BTreePageId nextp = curp.getRightSiblingId(); // 获取右兄弟页 ID

            if (nextp == null) {
                curp = null; // 没有更多页了
            } else {
                // 获取右兄弟页
                curp = (BTreeLeafPage) Database.getBufferPool().getPage(tid, nextp, Permissions.READ_ONLY);
                it = curp.iterator(); // 获取新的迭代器
                if (!it.hasNext()) // 如果新页也为空，则继续循环
                    it = null;
            }
        }

        // 返回下一个元组或 null
        if (it == null)
            return null;
        return it.next();
    }

    /**
     * 将迭代器重置到起始位置
     */
    public void rewind() throws DbException, TransactionAbortedException {
        close(); // 先关闭当前迭代器
        open(); // 再重新打开
    }

    /**
     * 关闭迭代器，释放资源
     */
    public void close() {
        super.close();
        it = null; // 清空迭代器引用
        curp = null; // 清空当前页引用
    }
}

/**
 * 辅助类，用于实现对 B+ 树文件中满足索引谓词条件的元组的迭代访问。
 * 支持基于索引的查询（如等于、大于、小于等操作）。
 */
class BTreeSearchIterator extends AbstractDbFileIterator {

    Iterator<Tuple>      it   = null; // 当前页的元组迭代器
    BTreeLeafPage        curp = null; // 当前正在遍历的叶子页

    final TransactionId  tid;        // 当前事务 ID
    final BTreeFile      f;          // 要遍历的 BTree 文件
    final IndexPredicate ipred;      // 索引过滤条件（谓词）

    /**
     * 构造函数
     *
     * @param f     - 包含数据的 BTreeFile
     * @param tid   - 当前事务 ID
     * @param ipred - 过滤条件（索引谓词）
     */
    public BTreeSearchIterator(BTreeFile f, TransactionId tid, IndexPredicate ipred) {
        this.f = f;
        this.tid = tid;
        this.ipred = ipred;
    }

    /**
     * 打开迭代器，从第一个符合条件的叶子页开始。
     * 根据索引谓词定位到合适的起始叶子页。
     */
    public void open() throws DbException, TransactionAbortedException {
        BTreeRootPtrPage rootPtr = (BTreeRootPtrPage) Database.getBufferPool().getPage(tid,
            BTreeRootPtrPage.getId(f.getId()), Permissions.READ_ONLY);
        BTreePageId root = rootPtr.getRootId();

        // 根据不同的谓词类型定位到对应的叶子页
        if (ipred.getOp() == Op.EQUALS || ipred.getOp() == Op.GREATER_THAN || ipred.getOp() == Op.GREATER_THAN_OR_EQ) {
            curp = f.findLeafPage(tid, root, ipred.getField());
        } else {
            // 对于 LESS_THAN 或其他情况，从最左叶子页开始查找
            curp = f.findLeafPage(tid, root, null);
        }

        it = curp.iterator(); // 初始化当前页的元组迭代器
    }

    /**
     * 读取下一个满足谓词条件的元组。
     * 若当前页无更多匹配项，则跳转到右兄弟页继续查找。
     *
     * @return 下一个符合条件的元组，若不存在则返回 null
     */
    @Override
    protected Tuple readNext() throws TransactionAbortedException, DbException, NoSuchElementException {
        while (it != null) {

            while (it.hasNext()) {
                Tuple t = it.next();

                // 检查当前元组是否满足谓词条件
                if (t.getField(f.keyField()).compare(ipred.getOp(), ipred.getField())) {
                    return t; // 返回符合条件的元组
                }

                // 如果是 LESS_THAN 或 EQUALS 操作且不满足条件，提前终止搜索
                else if (ipred.getOp() == Op.LESS_THAN || ipred.getOp() == Op.LESS_THAN_OR_EQ) {
                    return null;
                } else if (ipred.getOp() == Op.EQUALS
                           && t.getField(f.keyField()).compare(Op.GREATER_THAN, ipred.getField())) {
                    return null;
                }
            }

            // 当前页已读完，尝试读取右兄弟页
            BTreePageId nextp = curp.getRightSiblingId();
            if (nextp == null) {
                return null; // 没有更多兄弟页，结束迭代
            } else {
                curp = (BTreeLeafPage) Database.getBufferPool().getPage(tid, nextp, Permissions.READ_ONLY);
                it = curp.iterator(); // 更新为新页的迭代器
            }
        }

        return null; // 所有页都处理完毕，没有更多结果
    }

    /**
     * 将迭代器重置回初始状态，重新开始搜索
     */
    public void rewind() throws DbException, TransactionAbortedException {
        close(); // 先关闭当前迭代器
        open(); // 再重新打开
    }

    /**
     * 关闭迭代器，释放资源
     */
    public void close() {
        super.close();
        it = null; // 清空迭代器引用
    }
}
